package com.demo.rocketmq.sender;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

/**
 * 发送异步消息
 */
@Slf4j
@Configuration
public class AsyncMqSender {

    @Autowired
    private DefaultMQProducer defaultMQProducer;

    public void sendMsg(String topic, String msg, String tags, String keys, int delayLevel) throws RemotingException, InterruptedException, MQClientException {
        log.info("MQ发送异步消息，topic = {}，tags = {}，keys = {}，delayLevel = {}，msg = {}", topic, tags, keys, delayLevel, msg);
        Message message = new Message(topic, tags, keys, msg.getBytes());
        if (delayLevel > 0) {
            message.setDelayTimeLevel(delayLevel);
        }
        defaultMQProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
                    log.info("MQ发送异步消息成功，sendResult = {}", sendResult);
                } else {
                    log.error("MQ发送异步消息失败，sendResult = {}", sendResult);
                }
            }

            @Override
            public void onException(Throwable e) {
                log.error("MQ发送异步消息异常，e = {}", ExceptionUtils.getStackTrace(e));
                // 补偿机制，根据业务情况进行使用，看是否进行重试
            }
        });
    }


}
